// -*-mode:c++; coding:utf-8-*-

#ifndef _XBASE_GROUP_SCANNER_HPP_
#define _XBASE_GROUP_SCANNER_HPP_

#include <string>
#include <vector>
#include <queue>

#include <boost/shared_ptr.hpp>

#include "kv_scanner.hpp"

namespace xbase {
	// composite for group
	class group_scanner
		: public kv_scanner
	{
	public:
		typedef boost::shared_ptr<group_scanner> pointer;

		// 对SSTables和memory_table的混合体来说，Key相等的记录，
		// 一定先取最新写入的。所以加入 sequence_number 这一因
		// 素，来保证这一点。
		struct greater_compare {
			inline
			bool operator() (const kv_scanner::pointer &first,
					 const kv_scanner::pointer &second) const {
				key_equal_to key_eq;
				// second.entry.key < first.entry.key => first > second
				if(key_eq(*second->get().m_key,
					  *first->get().m_key)) // equal
				{
					// 作为时间戳来比较，新的（大的）是小的
					timestamp_less_than ts_lt;
					return ts_lt(second->sequence_number(),
						     first->sequence_number());
				}
				else // not equal
				{
					key_less_than key_lt;
					return key_lt(*second->get().m_key,
						      *first->get().m_key);
				}
			}
		};
		
		struct record_state
		{
			enum inner_type
			{
				IT_NO,
				IT_ROW,
				IT_FAMILY,
				IT_QUALIFIER
			};

			const bool m_filter_deleted_record_enabled;

			const bool m_filter_deletion_record_enabled;
			bool m_is_newest_deletion_record;

			inner_type m_deletion_type;

			const bool m_filter_version_enabled;
			size_t m_version_count;

			const bool m_need_save_status; // = ((m_filter_version_enabled)
						       // || (m_filter_deleted_record_enabled))
			std::string m_prev_row;
			std::string m_prev_family;
			std::string m_prev_qualifier;
			uint64_t m_prev_timestamp;
			inner_type m_change_type;
			bool m_is_duplicate_record; // 这一条记录与上一
						    // 条完全一样，在版
						    // 本数过滤开启状态
						    // 下应该被无条件忽
						    // 略，并且不占用版
						    // 本数。以实现通过
						    // 多次写同一个时间
						    // 戳的记录来修改这
						    // 条记录的功能
			bool m_duplicate_record_deleted; // 重复记录中有一个是删除记录，后面的重复记录应该被它删除了；但删除功能只限于重复记录中！
							 // 专门用于重复记录内部的删除

			std::string to_string() const;

			record_state(bool filter_deleted_record_enabled,
				     bool filter_deletion_record_enabled,
				     bool filter_version_enabled)
				: m_filter_deleted_record_enabled(filter_deleted_record_enabled)
				, m_filter_deletion_record_enabled(filter_deletion_record_enabled)
				, m_is_newest_deletion_record(false)
				, m_deletion_type(IT_NO)
				, m_filter_version_enabled(filter_version_enabled)
				, m_version_count(0)
				, m_need_save_status(filter_version_enabled || filter_deleted_record_enabled)
				, m_prev_timestamp(BAD_TIMESTAMP)
				, m_change_type(IT_NO)
				, m_is_duplicate_record(false)
				, m_duplicate_record_deleted(false) {}

			void update(const block_entry &entry,
				    const column_name_parser &parser);
		};

	public:
		group_scanner(const kv_scan_context::pointer &context)
			: kv_scanner(context)
			, m_state(context->filter_deleted_enabled(),
				  context->filter_deleted_itself_enabled(),
				  context->need_check_version()) {}

		virtual
		~group_scanner() {}

		virtual
		bool eos() const {
			return m_queue.empty();
		}

		virtual
		const block_entry &get() const {
			assert(! m_queue.empty());
			return m_queue.top()->get();
		}

		virtual
		void next() {
			// 先去掉上一个结果
			if(! m_queue.empty())
			{
				kv_scanner::pointer scanner = m_queue.top();
				m_queue.pop();
				scanner->next();
				const bool eos = scanner->eos();
				if(! eos)
				{
					m_queue.push(scanner);
				}
				// else drop it
			}

			fetch_entry(); // 再找下一条
		}

		void add(const kv_scanner::pointer &scanner) {
			if(! scanner->eos())
			{
				m_queue.push(scanner);
			}
			// else, drop it
		}

		// 当全部子扫描器加入完成后，调用init，来为读做准备
		void init() {
			fetch_entry();
		}

		bool empty() const {
			return m_queue.empty();
		}

		size_t size() const {
			return m_queue.size();
		}

	protected:
		inline
		void pop_first() {
			kv_scanner::pointer scanner = m_queue.top();
			m_queue.pop();
			scanner->next();
			const bool eos = scanner->eos();
			if(! eos)
			{
				m_queue.push(scanner);
			}
			// else drop it
		}

		// 在这里过滤时间戳、删除记录、版本数等等
		void fetch_entry();
	private:
		// make it a min-heap by greater_compare
		std::priority_queue<kv_scanner::pointer,
				    std::vector<kv_scanner::pointer>,
				    greater_compare> m_queue;

		record_state m_state;
	};

} // namespace xbase

#endif	// _XBASE_GROUP_SCANNER_HPP_
